# -*- coding: utf-8 -*-
"""
Project: automation-api-project
Creator: jinsheng
Create time: 2021-11-12 11:29
IDE: PyCharm
Introduction:
"""

import json
from datetime import datetime, timedelta

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
from pykafka import KafkaClient


class OperateKafka:
    def __init__(self, version=17):
        if version == 17:
            self.bootstrap_servers = ['10.0.17.17:9091', '10.0.17.17:9092', '10.0.17.17:9093']
            self.hosts = '10.0.17.17:9092'
        elif version == 221:
            self.bootstrap_servers = ['221.226.153.90:19091', '221.226.153.90:19092', '221.226.153.90:19093']
            self.hosts = '221.226.153.90:2181'
        elif version == 156:
            # self.bootstrap_servers = ['117.88.45.156:9092', '117.88.45.79:9092', '117.88.47.79:9092']
            self.bootstrap_servers = ['cdh01:9092', 'cdh02:9092', 'cdh03:9092']
            self.hosts = '117.88.45.156:2181'
        self.topic = None

    def send_message(self, topic, message):
        """
        向kafka指定topic发送消息
        @param topic: 指定的topic
        @param message: 发送的消息体(字典形式或字符串形式)
        @return: None
        """
        producer = KafkaProducer(bootstrap_servers=self.bootstrap_servers)
        send_message = None
        try:
            if isinstance(message, dict) or isinstance(message, list):
                send_message = json.dumps(message).encode()
            elif isinstance(message, str):
                send_message = message.encode()
            producer.send(topic, send_message)
        except Exception as e:
            print(e)
        finally:
            producer.close()

    def read_message(self, topic):
        """
        读取topic中的offset和value
        @param topic: 需要读取的topic
        @return: offset值和对应的value值
        """
        res_list = {}
        client = KafkaClient(hosts=self.hosts)
        topic_read = client.topics[topic]
        consumer = topic_read.get_simple_consumer(reset_offset_on_start=True)
        for message in consumer:
            if message:
                # return message.offset, message.value.decode()
                res_list[message.offset] = message.value.decode()
        return res_list

    def consumer_demo(self):
        # time_start = datetime.now()
        count = 0
        cs = KafkaConsumer('COMMUNICATION_RESPONSE_CONNECTION', bootstrap_servers=self.bootstrap_servers)
        for message in cs:
            res_str = message.value.decode()
            res2 = res_str.replace('true', 'True')
            res_list = eval(res2)
            success_code = res_list[0]['responseResults'][0]['deviceResults'][0]['data']['event_type']
            if success_code == 1:
                count += 1
            if count == 100:
                time_start = datetime.now()
            if count % 10000 == 0:
                time_end = datetime.now()
                time_res = time_end - time_start
                print(f'增加了{count}个，用时{time_res}秒')


if __name__ == '__main__':
    # ok = OperateKafka(version=156)
    # # ok.send_message('EVENT_PUSH_SERVICE_EVENT_PUSH', data)
    # fmt_s = '%Y%m%d%H%M%S'
    # count = 0
    # for j in range(0 + 1, 0 + 11):
    #     # for j in range(96):
    #     # device_id = str(100000000000000000 + j)
    #     device_id = str(10000000 + j)
    #     # device_id = '121926203731877888'
    #     tm = datetime(2022, 10, 1, 0, 0)
    #     for i in range(2880):
    #         tm_res = str(tm.strftime(fmt_s))[::-1]
    #         row_key = tm_res + device_id + 'mi15'
    #         data = {"indexColumns": "device_id,data_time,frequency", "rowkey": row_key,
    #                 "tableName": "ods_device_electric_ts"}
    #         ok.send_message('kafka.index.flink', data)
    #         tm += timedelta(minutes=15)
    #         count += 1
    #         print(count)
    # # ok.send_message('COMMUNICATION_RESPONSE_ANALOG', data)
    ok = OperateKafka(version=156)
    ok.consumer_demo()
